package com.yanggu.bigdata.realtime.app.dws

import cn.hutool.core.collection.CollUtil
import cn.hutool.core.date.DateUtil
import cn.hutool.core.thread.NamedThreadFactory
import cn.hutool.core.util.StrUtil
import cn.hutool.json.JSONUtil
import com.alibaba.fastjson.JSON
import com.yanggu.bigdata.realtime.bean.{OrderWide, PaymentWide, ProductStats}
import com.yanggu.bigdata.realtime.common.GmallConfig.KAFKA_BROKER_LIST
import com.yanggu.bigdata.realtime.common.GmallConstant
import com.yanggu.bigdata.realtime.utils.{ClickHouseUtil, DimQueryUtil, KafkaUtil}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.async.{AsyncFunction, ResultFuture}
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.time.Duration
import java.util.concurrent.{ArrayBlockingQueue, CompletableFuture, ThreadPoolExecutor, TimeUnit}
import java.util.{Collections, Date}
import java.{lang, util}

/**
 * Desc: 形成以商品为准的统计 曝光 点击 购物车 下单 支付 退单 评论数 宽表
 * 数据流(行为数据) client -> nginx -> springboot -> kafka(ods_base_log) -> FlinkApp(ods to dwd) -> Kafka(dwd_xxx_log) -> FlinkApp(ods to dwm) -> Kafka(dwm_xxx)
 *                                                                                                                                                                        -> FlinkApp(dwd、dwm to dws) -> clickhouse
 * 数据流(业务数据) client -> nginx -> springboot -> kafka(ods_base_db) -> FlinkApp(ods to dwd) -> Kafka(dwd_xxx_info)、hbase(dim_) -> FlinkApp(ods to dwm) -> kafka(dwm_xxx)
 *
 * 程序(行为数据) logger.sh(mock-log) -> nginx -> springboot -> kafka(ods_base_log) -> FlinkApp(BaseLogApp) -> kafka(dwd_xxx_log) -> FlinkApp(UniqueVisitApp)
 *                                                                                                                                                                           -> FlinkApp(ProductStatsApp) -> clickhouse(product_stats)
 * 程序(业务数据) db.sh(mock-db) -> MySQL -> FlinkApp(CDC) -> kafka(ods_base_db) -> FlinkApp(BaseDBApp) -> kafka(dwd_xx)、hbase(dim) -> FlinkApp(OrderWideApp、PaymentWideApp)
 */
object ProductStatsApp {

  def main(args: Array[String]): Unit = {
    //1. 创建执行环境
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    //2. 把json字符串数据流转换成统一数据对象的数据流
    val groupId = "product_stats_app"
    val pageViewSourceTopic = "dwd_page_log"
    val orderWideSourceTopic = "dwm_order_wide"
    val favorInfoSourceTopic = "dwd_favor_info"
    val cartInfoSourceTopic = "dwd_cart_info"
    val paymentWideSourceTopic = "dwm_payment_wide"
    val refundInfoSourceTopic = "dwd_order_refund_info"
    val commentInfoSourceTopic = "dwd_comment_info"

    //2.1 转换曝光及页面流数据
    val pageAndDisplayDataStream = KafkaUtil.getKafkaDataStream(environment, KAFKA_BROKER_LIST, pageViewSourceTopic, groupId)
      .flatMap((value: String, out: Collector[ProductStats]) => {
        val jsonObj = JSON.parseObject(value)
        val pageObj = jsonObj.getJSONObject("page")
        val pageId = pageObj.getString("page_id")
        val ts = jsonObj.getLong("ts")
        if (StrUtil.equals(pageId, "good_detail")) {
          val skuId = pageObj.getLong("item")
          val stats = ProductStats.builder()
            .sku_id(skuId)
            .click_ct(1L)
            .ts(ts)
            .build()
          out.collect(stats)
        }
        val displays = jsonObj.getJSONArray("display")
        if (CollUtil.isNotEmpty(displays)) {
          val length = displays.size()
          for (index <- 0 until length) {
            val display = displays.getJSONObject(index)
            if (StrUtil.equals("sku_id", display.getString("item_type"))) {
              val skuId = display.getLong("item")
              val productStats = ProductStats.builder()
                .sku_id(skuId)
                .display_ct(1L)
                .build()
              out.collect(productStats)
            }
          }
        }
      })

    //2.2 下单数据流
    val orderWideDataStream = KafkaUtil.getKafkaDataStream(environment, KAFKA_BROKER_LIST, orderWideSourceTopic, groupId)
      .map(json => {
        val orderWide = JSONUtil.toBean(json, classOf[OrderWide])
        ProductStats.builder()
          .sku_id(orderWide.getSku_id)
          .orderIdSet(new util.HashSet[lang.Long](Collections.singleton(orderWide.getOrder_id)))
          .order_sku_num(orderWide.getSku_num)
          .order_amount(orderWide.getSplit_total_amount)
          .ts(DateUtil.parseDateTime(orderWide.getCreate_time).getTime)
          .build()
      })

    //2.3 收藏数据流
    val favorDataStream = KafkaUtil.getKafkaDataStream(environment, KAFKA_BROKER_LIST, favorInfoSourceTopic, groupId)
      .map(json => {
        val data = JSON.parseObject(json)
        ProductStats.builder()
          .sku_id(data.getLong("sku_id"))
          .favor_ct(1L)
          .ts(DateUtil.parseDateTime(data.getString("create_time")).getTime)
          .build()
      })

    //2.4 购物车数据流
    val carInfoDataStream = KafkaUtil.getKafkaDataStream(environment, KAFKA_BROKER_LIST, cartInfoSourceTopic, groupId)
      .map(json => {
        val data = JSON.parseObject(json)
        ProductStats.builder()
          .sku_id(data.getLong("sku_id"))
          .cart_ct(1L)
          .ts(DateUtil.parseDateTime(data.getString("create_time")).getTime)
          .build()
      })

    //2.5 支付数据流
    val paymentDataStream = KafkaUtil.getKafkaDataStream(environment, KAFKA_BROKER_LIST, paymentWideSourceTopic, groupId)
      .map(json => {
        val data = JSONUtil.toBean(json, classOf[PaymentWide])
        ProductStats.builder()
          .sku_id(data.getSku_id)
          .payment_amount(data.getSplit_total_amount)
          .paidOrderIdSet(new util.HashSet[lang.Long](Collections.singleton(data.getOrder_id)))
          .ts(DateUtil.parseDateTime(data.getPayment_create_time).getTime)
          .build()
      })

    //2.6 退款数据流
    val refundDataStream = KafkaUtil.getKafkaDataStream(environment, KAFKA_BROKER_LIST, refundInfoSourceTopic, groupId)
      .map(json => {
        val data = JSON.parseObject(json)
        ProductStats.builder()
          .sku_id(data.getLong("sku_id"))
          .refund_amount(data.getBigDecimal("refund_amount"))
          .refundOrderIdSet(new util.HashSet[lang.Long](Collections.singleton(data.getLong("order_id"))))
          .ts(DateUtil.parseDateTime(data.getString("create_time")).getTime)
          .build()
      })

    //2.7 评价数据流
    val commentDataStream = KafkaUtil.getKafkaDataStream(environment, KAFKA_BROKER_LIST, commentInfoSourceTopic, groupId)
      .map(json => {
        val data = JSON.parseObject(json)
        ProductStats.builder()
          .sku_id(data.getLong("sku_id"))
          .comment_ct(1L)
          .good_comment_ct(if (StrUtil.equals(data.getString("appraise"), GmallConstant.APPRAISE_GOOD)) 1L else 0L)
          .ts(DateUtil.parseDateTime(data.getString("create_time")).getTime)
          .build()
      })

    //3. 把多个流的数据进行union合并成同一个流
    val productStatsDataStream = pageAndDisplayDataStream.union(orderWideDataStream, carInfoDataStream, paymentDataStream, refundDataStream, favorDataStream, commentDataStream)

      //4. 设置watermark和延迟时间
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forBoundedOutOfOrderness[ProductStats](Duration.ofSeconds(1L))
        .withTimestampAssigner(new SerializableTimestampAssigner[ProductStats] {
          override def extractTimestamp(element: ProductStats, recordTimestamp: Long): Long = element.getTs
        })
      )
      //5. 分组、开窗、聚合
      .keyBy(data => data.getSku_id)
      .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
      .allowedLateness(Time.seconds(1L))
      .reduce(new ReduceFunction[ProductStats] {
        override def reduce(value1: ProductStats, value2: ProductStats): ProductStats = {
          //曝光数
          value1.setDisplay_ct(value2.getDisplay_ct + value1.getDisplay_ct)
          //点击数
          value1.setClick_ct(value2.getClick_ct + value1.getClick_ct)
          //收藏数
          value1.setFavor_ct(value2.getFavor_ct + value1.getFavor_ct)
          //添加购物车数
          value1.setCart_ct(value2.getCart_ct + value1.getCart_ct)
          //下单商品个数
          value1.setOrder_sku_num(value2.getOrder_sku_num + value1.getOrder_sku_num)
          //下单商品金额
          value1.setOrder_amount(value2.getOrder_amount.add(value1.getOrder_amount))
          //支付金额
          value1.setPayment_amount(value2.getPayment_amount.add(value1.getPayment_amount))
          //退款金额
          value1.setRefund_amount(value2.getRefund_amount.add(value1.getRefund_amount))
          //评价订单数
          value1.setComment_ct(value2.getComment_ct + value1.getComment_ct)
          //好评订单数
          value1.setGood_comment_ct(value2.getGood_comment_ct + value1.getGood_comment_ct)

          //下单订单id的Set
          value1.getOrderIdSet.addAll(value2.getOrderIdSet)
          //支付订单id的Set
          value1.getPaidOrderIdSet.addAll(value2.getPaidOrderIdSet)
          //退款订单id的Set
          value1.getRefundOrderIdSet.addAll(value2.getRefundOrderIdSet)

          //ts
          value1.setTs(Math.max(value1.getTs, value2.getTs))

          value1
        }
      },
        new ProcessWindowFunction[ProductStats, ProductStats, java.lang.Long, TimeWindow] {
          override def process(key: lang.Long, context: Context, elements: Iterable[ProductStats], out: Collector[ProductStats]): Unit = {
            val data = elements.head

            //订单数
            data.setOrder_ct(data.getOrderIdSet.size())

            //支付订单数
            data.setPaid_order_ct(data.getPaidOrderIdSet.size())

            //退款订单数
            data.setRefund_order_ct(data.getRefundOrderIdSet.size())

            //补充窗口时间字段
            val window = context.window
            data.setStt(DateUtil.formatDateTime(new Date(window.getStart)))
            data.setEdt(DateUtil.formatDateTime(new Date(window.getEnd)))

            out.collect(data)
          }
        }
      )

    //6. 补充维度数据
    val resultDatStream = AsyncDataStream.unorderedWait(productStatsDataStream, new AsyncFunction[ProductStats, ProductStats] {

      private lazy val executor = {
        val threadPoolExecutor = new ThreadPoolExecutor(10, 20, 100L, TimeUnit.SECONDS,
          new ArrayBlockingQueue[Runnable](100), new NamedThreadFactory("add-dim-data", false),
          new ThreadPoolExecutor.DiscardPolicy())
        //直接初始化所有核心线程
        threadPoolExecutor.prestartAllCoreThreads()
        threadPoolExecutor
      }

      override def asyncInvoke(input: ProductStats, resultFuture: ResultFuture[ProductStats]): Unit = {
        val list = new util.ArrayList[CompletableFuture[Void]](6)

        //SKU维度信息补充
        val skuFuture = CompletableFuture.runAsync(() => {
          val dimInfo = DimQueryUtil.getDimInfo("DIM_SKU_INFO", pkValue = input.getSku_id.toString)
          if (dimInfo == null) {
            return
          }
          input.setSku_name(dimInfo.getString("SKU_NAME"))
          input.setSku_price(dimInfo.getBigDecimal("PRICE"))
          input.setSpu_id(dimInfo.getLong("SPU_ID"))
          input.setCategory3_id(dimInfo.getLong("CATEGORY3_ID"))
          input.setTm_id(dimInfo.getLong("TM_ID"))
        }, executor)
        list.add(skuFuture)

        //SPU维度信息补充
        val spuFuture = skuFuture.thenRunAsync(() => {
          if (input.getSpu_id == null) {
            return
          }
          val dimInfo = DimQueryUtil.getDimInfo("DIM_SPU_INFO", pkValue = input.getSpu_id.toString)
          if (dimInfo == null) {
            return
          }
          input.setSpu_name(dimInfo.getString("SPU_NAME"))
        }, executor)
        list.add(spuFuture)

        //品牌维度信息补充
        val tmFuture = skuFuture.thenRunAsync(() => {
          if (input.getTm_id == null) {
            return
          }
          val dimInfo = DimQueryUtil.getDimInfo("DIM_BASE_TRADEMARK", pkValue = input.getTm_id.toString)
          if (dimInfo == null) {
            return
          }
          input.setTm_name(dimInfo.getString("TM_NAME"))
        }, executor)
        list.add(tmFuture)

        //3级分类维度信息补充
        val category3Future = skuFuture.thenRunAsync(() => {
          if (input.getCategory3_id == null) {
            return
          }
          val dimInfo = DimQueryUtil.getDimInfo("DIM_BASE_CATEGORY3", pkValue = input.getCategory3_id.toString)
          if (dimInfo == null) {
            return
          }
          input.setCategory3_name(dimInfo.getString("NAME"))
        }, executor)
        list.add(category3Future)

        val array = new Array[CompletableFuture[Void]](list.size())

        //scala调用java的可变参数
        //https://blog.csdn.net/eases_stone/article/details/85096585
        //等待所有线程执行完成后, 向下游传递数据或者异常
        CompletableFuture.allOf(list.toArray(array): _*)
          .whenCompleteAsync((_, exception) => {
            if (exception != null) {
              resultFuture.completeExceptionally(exception)
              return
            }
            resultFuture.complete(Iterable(input))
          })
      }
    }, 600L, TimeUnit.SECONDS, 1000)

    //7. 写入clickhouse
    resultDatStream.print("result>>>>>>>")
    resultDatStream.addSink(ClickHouseUtil.getSinkFunction[ProductStats]("product_stats", classOf[ProductStats]))

    //8. 启动job
    environment.execute("ProductStatsApp Job")

  }

}
